package hbasecli.controller;

import hbasecli.exception.ConvertException;
import hbasecli.hbase.HbaseDao;
import hbasecli.hbase.HbaseRecord;
import hbasecli.hbase.HbaseRecordInput;
import hbasecli.hbase.HbaseRecordOutput;
import hbasecli.param.CreateRecord;
import hbasecli.param.DeleteRecord;
import hbasecli.param.ListRecord;
import hbasecli.resp.Resp;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

import static cn.hutool.core.collection.CollUtil.isEmpty;
import static cn.hutool.core.util.StrUtil.isBlankIfStr;
import static hbasecli.resp.Resp.failure;
import static hbasecli.resp.Resp.success;

@RestController
public class RecordController {

    @Resource
    private HbaseDao hbaseDao;

    @PostMapping("/api/listRecord")
    public Resp<List<HbaseRecordOutput>> listRecord(@RequestBody ListRecord param) {
        String tableName = param.getTableName();
        if (isBlankIfStr(tableName)) {
            return failure("表名不可以为空");
        }
        String rowKeyPattern = param.getRowKeyPattern();
        List<HbaseRecord> hbaseRecords;
        if (isBlankIfStr(rowKeyPattern)) {
            hbaseRecords = hbaseDao.scan(tableName);
        } else {
            hbaseRecords = hbaseDao.fuzzyScan(tableName, rowKeyPattern, '_');
        }
        return success(hbaseRecords.stream().map(HbaseRecord::toOutput).collect(Collectors.toList()));
    }

    @PostMapping("/api/createRecord")
    public Resp<Boolean> createRecord(@RequestBody CreateRecord param) {
        String tableName = param.getTableName();
        if (isBlankIfStr(tableName)) {
            return failure("表名不可以为空");
        }
        List<HbaseRecordInput> recordInputs = param.getRecords();
        if (isEmpty(recordInputs)) {
            return failure("至少需要一条数据");
        }
        for (HbaseRecordInput recordInput : recordInputs) {
            if (isBlankIfStr(recordInput.getRowKey())) {
                return failure("行键不可以为空");
            }
            if (isBlankIfStr(recordInput.getRowKeyType())) {
                return failure("行键需要指定存储类型");
            }
            if (isEmpty(recordInput.getCol2val())) {
                return failure("至少需要一个列值");
            }
            for (String col : recordInput.getCol2val().keySet()) {
                String[] strings = col.split(":");
                if (strings.length != 3
                        || isBlankIfStr(strings[0])
                        || isBlankIfStr(strings[1])
                        || isBlankIfStr(strings[2])) {
                    throw new ConvertException("参数格式出错,请检查列族,列限定符或数据类型");
                }
            }
        }
        boolean ok = true;
        for (HbaseRecordInput recordInput : recordInputs) {
            HbaseRecord record = recordInput.toRecord();
            List<HbaseRecord.HbaseRecordColumn> columns = record.getColumns();
            for (HbaseRecord.HbaseRecordColumn column : columns) {
                ok &= hbaseDao.put(
                        tableName,
                        record.getRowKey(),
                        column.getFamily(),
                        column.getQualifier(),
                        column.getValue());
            }
        }
        return success(ok);
    }

    @PostMapping("/api/deleteRecord")
    public Resp<Boolean> deleteRecord(@RequestBody DeleteRecord param) {
        String tableName = param.getTableName();
        if (isBlankIfStr(tableName)) {
            return failure("表名不可以为空");
        }
        List<List<Byte>> rowKeys = param.getRowKeys();
        if (isEmpty(rowKeys)) {
            return failure("rowKeys不可以为空");
        }
        List<byte[]> rowKeyList = new ArrayList<>();
        for (List<Byte> byteList : rowKeys) {
            int length = byteList.size();
            byte[] bytes = new byte[length];
            for (int i = 0; i < length; i++) {
                bytes[i] = byteList.get(i);
            }
            rowKeyList.add(bytes);
        }
        return success(hbaseDao.deleteRecord(tableName, rowKeyList));
    }
}
